
import com.aliyun.odps.data.Record;
        import com.aliyun.odps.data.TableInfo;
        import com.aliyun.odps.mapred.JobClient;
        import com.aliyun.odps.mapred.MapperBase;
        import com.aliyun.odps.mapred.ReducerBase;
        import com.aliyun.odps.mapred.conf.JobConf;
        import com.aliyun.odps.mapred.utils.InputUtils;
        import com.aliyun.odps.mapred.utils.OutputUtils;
        import com.aliyun.odps.mapred.utils.SchemaUtils;

        import java.io.IOException;
import java.text.ParseException;
import java.util.Iterator;

public class mileage_fault {
    public static class TokenizerMapper extends MapperBase {
        private Record key;
        private Record value;


        @Override
        public void setup(TaskContext context) throws IOException {
            key= context.createMapOutputKeyRecord();
            value = context.createMapOutputValueRecord();

        }
        @Override
        public void map(long recordNum, Record record, TaskContext context)
                throws IOException {
            VIN vin1 = new VIN(record.getString("vin"));
            String year = vin1.getResult().split(":")[2];
            String diagnosis_year="空";
            try {
                diagnosis_year = util.Date_year(record.getDatetime("diagnosis_time"));
            } catch (ParseException e) {
                e.printStackTrace();
            }
            String mileage = mileage_regex.mileage(record.getString("mileage")).toString();
            String province = util.province_notnull(record.getString("province"));
            String fault_description = new Res_fault().code_descroption(record.getString("fault_codes"));
            String[] fault_descriptionArr = fault_description.split("QAZ123");
            String displacement = util.LAT_notnull(record.getString("displacement"));

            for(int i=0;i<fault_descriptionArr.length;i++){
                key.set("year",year);
                key.set("diagnosis_year", diagnosis_year);
                key.set("mileage",mileage);
                key.set("province",province);
                key.set("displacement",displacement);
                if(fault_descriptionArr[i].split("qaz123").length==2) {
                    key.set("fault", fault_descriptionArr[i].split("qaz123")[0]);
                    value.set("fault_description", fault_descriptionArr[i].split("qaz123")[1]);

                }
                value.set("fault_sum",1L);
                value.set("car_sum",1L);
                context.write(key, value);

            }




        }
    }


    public static class SumCombiner extends ReducerBase {
        private Record count;
        @Override
        public void setup(TaskContext context) throws IOException {
            count = context.createMapOutputValueRecord();
        }
        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
                throws IOException {
            long c = 0;
            long d = 0;
            String s1 = "";
            String s2 = "";

            while (values.hasNext()) {
                Record val = values.next();
                s2=(String) val.get(0);
                if(s2==null){
                    s2 = "";
                }

                if(s1.length()<s2.length()){
                    s1 = s2;
                }


                c += (Long) val.get(1);
                d += (Long) val.get(2);
            }
            count.set(0, s1);
            count.set(1, c);
            count.set(2, d);
            context.write(key, count);
        }

    }

    public static class SumReducer extends ReducerBase {
        private Record result = null;
        @Override
        public void setup(TaskContext context) throws IOException {
            result = context.createOutputRecord();
        }
        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
                throws IOException {
            long c = 0;
            long d = 0;
            String s1 = "";
            String s2 = "";

            while (values.hasNext()) {
                Record val = values.next();
                s2=(String) val.get(0);
                if(s2==null){
                    s2 = "";
                }

                if(s1.length()<s2.length()){
                    s1 = s2;
                }


                c += (Long) val.get(1);
                d += (Long) val.get(2);
            }

            result.set(0, key.get(0));
            result.set(1, key.get(1));

            result.set(2, key.get(2));
            result.set(3, key.get(3));
            result.set(4, key.get(4));
            result.set(5, key.get(5));
            result.set(6, s1);
            result.set(7, c);
            result.set(8, d);
            context.write(result);
        }
    }
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: WordCount <in_table> <out_table>");
            System.exit(2);
        }
        JobConf job = new JobConf();
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(SumCombiner.class);
        job.setReducerClass(SumReducer.class);
        job.setMapOutputKeySchema(SchemaUtils.fromString("year:string,diagnosis_year:string,mileage:string,province:string,displacement:string,fault:string"));
        job.setMapOutputValueSchema(SchemaUtils.fromString("fault_description:string,fault_sum:bigint,car_sum:bigint"));
        InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
        JobClient.runJob(job);
    }
}
